package com.jqk.procuder.listener;

import com.alibaba.fastjson.JSON;
import com.jqk.procuder.entity.Order;
import com.jqk.procuder.entity.TransactionLog;
import com.jqk.procuder.repository.TransactionLogRepository;
import com.jqk.procuder.service.OrderService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;

import java.util.Objects;

@Slf4j
@RocketMQTransactionListener
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class OrderTransactionListener implements RocketMQLocalTransactionListener {

    @Autowired
    OrderService orderService;

    @Autowired
    TransactionLogRepository transactionLogRepository;

    /**
     * 回查方法
     * 执行逻辑：如果 executeLocalTransaction() 返回给 broker 的状态是  UNKNOWN 或者 1分钟后未返回状态
     * Broker通知 Producer检查本地事务，在Producer中回调本地事务监听器中的checkLocalTransaction方法，
     * Broker 再根据此方法返回的状态判断是提交  half msg 还是回滚
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        MessageHeaders headers = message.getHeaders();
        //获取事务ID
        String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
        log.info("检查本地事务,事务ID:{}",transactionId);
        //根据事务id从日志表检索
        TransactionLog repository = transactionLogRepository.getById(transactionId);
        if(Objects.nonNull(repository)){
            return RocketMQLocalTransactionState.COMMIT;
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }

    /**
     * 发送half msg 返回send ok后调用的方法，执行本地事务，返回本地事务执行状态
     * 如果本地事务是 COMMIT,告诉 broker 可以提交 half msg,消息才可以被投递被消费
     * 如果本地事务是 ROLLBACK,告诉 broker 回滚掉 half msg
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {
        log.info("开始执行本地事务....");
        MessageHeaders headers = message.getHeaders();
        //获取事务ID
        String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
        Order order = (Order) headers.get("body");
        log.info("transactionId is {}, order is {}",transactionId, JSON.toJSONString(order));
        try{
            //执行本地事务，并记录日志
           return orderService.createOrder(order, transactionId) ? RocketMQLocalTransactionState.COMMIT:RocketMQLocalTransactionState.ROLLBACK;
        }catch (Exception e){
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

}